/** 
 * dbagent  
 * com.iqiyi.dbagent.contraller.impl  
 */
package com.iqiyi.dbagent.contraller.impl;

import java.io.BufferedReader;
import java.sql.Connection;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.iqiyi.dbagent.base.config.CommParme;
import com.iqiyi.dbagent.base.config.CommParme.DEALAFTER;
import com.iqiyi.dbagent.base.config.CommParme.INPUTTYPR;
import com.iqiyi.dbagent.base.config.i.Configuration;
import com.iqiyi.dbagent.base.i.IExcutorCallable;
import com.iqiyi.dbagent.base.i.ISqlMonitorServer;
import com.iqiyi.dbagent.base.impl.ISqlExcutorCallableImpl;
import com.iqiyi.dbagent.base.utils.text.MapUtil;
import com.iqiyi.dbagent.base.utils.text.StringUtil;
import com.iqiyi.dbagent.base.utils.time.DateUtil;
import com.iqiyi.dbagent.contraller.i.IDataStoreHandler;
import com.iqiyi.dbagent.model.DealAfter;
import com.iqiyi.dbagent.model.Table4QiData;
import com.iqiyi.dbagent.view.monitor.alarm.Alarm;
import com.iqiyi.dbagent.view.monitor.alarm.SelfAlarm;

/**
 * @Title: DataTransStoreHandler.java
 * @Package com.iqiyi.dbagent.contraller.impl
 * @Description: 批量数据使用事务存储
 * @author 王年明
 * @date 2015年11月21日 下午7:28:09
 * @version V1.0
 */
public class DataTransStoreHandler implements IDataStoreHandler {

	private static final int bufferSize = 50000;

	private static final int excutorPoolSize = 50;

	private List<Future<Object>> resultList = new ArrayList<Future<Object>>();

	private static List<String> bufferList = new ArrayList<String>();

	private ExecutorService pool = null;

	private static final Logger logger = LoggerFactory.getLogger(DataTransStoreHandler.class);

	private long startDealTime;

	private long endDealTime;

	private long todoSize = 0l;

	private long dealSize = 0l;

	private DealAfter after = null;

	public ExecutorService getPool() {
		return pool;
	}

	public void setPool(ExecutorService pool) {
		this.pool = pool;
	}

	public Long getStartDealTime() {
		return startDealTime;
	}

	public void setStartDealTime(Long startDealTime) {
		this.startDealTime = startDealTime;
	}

	public Long getEndDealTime() {
		return endDealTime;
	}

	public void setEndDealTime(Long endDealTime) {
		this.endDealTime = endDealTime;
	}

	public DataTransStoreHandler() {
		// Runtime.getRuntime().availableProcessors()
		pool = Executors.newFixedThreadPool(excutorPoolSize);
		after = new DealAfter();
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.iqiyi.dbagent.contraller.i.IDataStoreHandler#dealData()
	 */
	public Object dealData(String line) {

		return after;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * com.iqiyi.dbagent.contraller.i.IDataStoreHandler#dealDataChian(com.iqiyi
	 * .dbagent.contraller.i.IDataStoreHandler)
	 */
	public Object dealDataChian(IDataStoreHandler handler) {
		// TODO Auto-generated method stub
		return null;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.iqiyi.dbagent.contraller.i.IDataStoreHandler#dealData(java.io.
	 * BufferedReader)
	 */
    @SuppressWarnings("unchecked")
    public Object dealData(BufferedReader bfr, Configuration config, ISqlMonitorServer server) {
		String rline = "";
		Map<String, Table4QiData> tbMap = config.getTableMap();
		startDealTime = System.currentTimeMillis();
		int bakCount = 0;
		try {
			while ((rline = bfr.readLine()) != null && !rline.trim().equals("") && rline.length() > 3) {
				String sline = "";
				String busiType = rline.split(config.getBusiTypeFilter())[0];
				if (null != tbMap.get(busiType)) {
					sline = paseData(busiType, tbMap, rline.trim(),config.getInputType());
					++todoSize;
					if (null == bufferList) {
						bufferList = new ArrayList<String>();
					}
					if (null != bufferList) {
						if (bufferList.size() < bufferSize) {
						    if(sline!=null&&!sline.isEmpty()){
						        bufferList.add(sline);
						    }
						}
						if (bufferList.size() == bufferSize) {
							try {
								IExcutorCallable iec = new ISqlExcutorCallableImpl(bufferList,config);
								Future<Object> fut = pool.submit(iec);
								resultList.add(fut);
							} catch (Exception e) {
								e.printStackTrace();
							}
							bufferList = null;
						}
					}
				} else {
					logger.info("不合法的源数据:{}", rline);
					Alarm alarm = new Alarm(DateUtil.getDateTime(new Date()), "数据不符合解析规则", "检查数据->",rline);
					SelfAlarm.sendAlarm(alarm,null,null,"不合法的源数据:" + rline);
					after.setDealEndTime(new Date());
					return after;
				}
				sline = null;
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		if (null != bufferList && bufferList.size() > 0) {
			try {
				IExcutorCallable iec = new ISqlExcutorCallableImpl(bufferList,config);
				Future<Object> fut = pool.submit(iec);
				resultList.add(fut);
			} catch (Exception e) {
				e.printStackTrace();
				logger.error("SQL poll err:{}", e.getCause());
			}
		}
		pool.shutdown();
		// 关闭线程池
		Map<Connection,Integer> reMap = new ConcurrentHashMap<Connection, Integer>();
		for (Future<Object> f : resultList) {
			try {
			    Map<Connection,Integer> ireMap = (Map<Connection, Integer>) f.get();
				for(Connection conn:ireMap.keySet()){
                    if (null != conn){
				        dealSize += ireMap.get(conn);
				        reMap.put(conn, ireMap.get(conn));
				    }
				}
			} catch (InterruptedException e) {
				e.printStackTrace();
			} catch (ExecutionException e) {
				e.printStackTrace();
			}
		}
        if (null != reMap){
		    if (config.isUseTransaction()) {
		        logger.info("todoSize:{},dealSize:{}",todoSize,dealSize);
		        if (todoSize != dealSize) {
		            for (Connection conn : reMap.keySet()) {
		                try {
		                    conn.rollback();
		                    conn.close();
		                    ++ bakCount;
		                    logger.info("{},goback", conn);
		                } catch (Exception e) {
		                    e.printStackTrace();
		                }
		            }
		        }else {
		            for (Connection conn : reMap.keySet()) {
		                try {
		                    conn.commit();
		                    conn.close();
		                    logger.info("{},commit", conn);
		                } catch (Exception e) {
		                    e.printStackTrace();
		                }
		            }
		        }
		    }else {
		        for (Connection conn : reMap.keySet()) {
		            try {
		                conn.commit();
		                conn.close();
		                logger.info("{},commit", conn);
		            } catch (Exception e) {
		                e.printStackTrace();
		            }
		        }
		    }
		}
		endDealTime = System.currentTimeMillis();
		after.setDealStartTime(new Date(startDealTime));
		after.setDealEndTime(new Date(endDealTime));
		after.setTodoSize(todoSize);
		after.setDealSize(dealSize);
		StringBuffer busiInfo = new StringBuffer();
		for(Table4QiData tdata:tbMap.values()){
		    busiInfo.append("\r\n"+tdata.getTableTag() + "->" + tdata.getRecordCount());
		}
		logger.info("此次处理耗时:{} ms", (endDealTime - startDealTime));
		logger.info("共需处理:{}条,处理完成或回滚了:{}条", todoSize, dealSize);
		logger.info("详细:{}",busiInfo.toString());
		if(bakCount>0){
		    Alarm alarm = new Alarm(DateUtil.getDateTime(new Date()), "有异常数据,需回滚", "检查数据内容","回滚,数据库连接数,线程数:"+bakCount);
		    SelfAlarm.sendAlarm(alarm,null,null,"异常回滚");
		}else{
		    Alarm alarm = new Alarm(DateUtil.getDateTime(new Date()), "此次入库成功", "无异常!","此次入库:"+dealSize+"条!"+"耗时:"+(endDealTime - startDealTime)+" ms,其中: "+busiInfo.toString());
		    SelfAlarm.sendAlarm(alarm,null,null,"正常入库");
		}
		after.setNoticeInfo(new StringBuffer(MapUtil.viewMap("各业务类型处理数据量:",tbMap)));
		after.setDealAfter(DEALAFTER.DEALOK);
    	if (null != server){
            server.stopMonitor();
        }
		return after;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * com.iqiyi.dbagent.contraller.i.DBAgentTemplate#paseData(com.iqiyi.dbagent
	 * .base.config.i.Configuration, java.lang.String)
	 */
	public String paseData(String tag, Map<String, Table4QiData> tbMap,String iData,INPUTTYPR iType) {
	    String currentData = iData;
	    StringBuffer sbf = new StringBuffer();
		Table4QiData qdata = tbMap.get(tag);
		qdata.setRecordCount(qdata.getRecordCount()+1);
		String tabId= qdata.getTabId();
		iData = iData.replace(qdata.getTagFilter(), "").replace(qdata.getTableTag(), "");
		iData = iData.replace("\01", ",");
		iData = iData.replace("\t", ",").trim();
		String ivalues[] = iData.split(",");
		iData = iData.replace(ivalues[0]+",,", ivalues[0]+",");
		String values[] = iData.split(",");
		List<String> filedList = qdata.getFiledsList();
		if(values.length !=filedList.size()){
		    logger.error("字段值:{}列数不够",currentData);
		    return null;
		}
		Map<String, Double> vMap = qdata.getValuesMap();
		sbf.append("insert into ");
		String tbName = getItabName(qdata.getSrcTableName(),Long.parseLong(values[0])*1000L);
		sbf.append(tbName);
		sbf.append("(");
        for (int i = 0; i < filedList.size(); i++) {
			String filed = filedList.get(i);
            sbf.append(filed + ",");
            if (null != vMap.get(filed)){
                try {
                    vMap.put(filed, Double.parseDouble(values[i]));
                } catch (Exception e) {
                    e.printStackTrace();
                    logger.error("类型:{},数据:{}异常,字段没有值,或不是数值类型!",tag,iData);
                }
            }
		}
		sbf = new StringBuffer(sbf.substring(0, sbf.lastIndexOf(",")));
		if(null != tabId && !"".equals(tabId)){
		    sbf.append(",id");
		}
		sbf.append(")values(");
		for (String value : values) {
			sbf.append("'" + value.trim() + "',");
		}
		sbf = new StringBuffer(sbf.substring(0, sbf.lastIndexOf(",")));
        if (null != tabId && !"".equals(tabId)){
            String indexs[] = StringUtil.getMiddStr(qdata.getTabId(),"${","}$").split(",");
            StringBuffer ids = new StringBuffer();
            for (String index : indexs){
                ids.append(values[Integer.parseInt(index)]).append("|");
            }
            sbf.append(",'"+ids.toString()+"'");
		}
		sbf.append(")");
        switch (iType) {
            case DEFAULT:
                break;
            case INSERT:
                break;
            case INCREASE:
                sbf.append("ON DUPLICATE KEY UPDATE ");
                for(String fKey:vMap.keySet()){
                    sbf.append(fKey+"="+fKey+"+"+vMap.get(fKey)+","); 
                }
                sbf = new StringBuffer(sbf.substring(0, sbf.lastIndexOf(",")));
                break;
            case REDUCE:
                sbf.append("ON DUPLICATE KEY UPDATE ");
                for(String fKey:vMap.keySet()){
                    sbf.append(fKey+"="+fKey+"-"+vMap.get(fKey)+","); 
                }
                sbf = new StringBuffer(sbf.substring(0, sbf.lastIndexOf(",")));
                break;
            case UPDATE:
                sbf.append("ON DUPLICATE KEY UPDATE ");
                for(String fKey:vMap.keySet()){
                    sbf.append(fKey+"="+vMap.get(fKey)+","); 
                }
                sbf = new StringBuffer(sbf.substring(0, sbf.lastIndexOf(",")));
                break;
            default:
                break;
        }
		//logger.info(">:{}",sbf.toString());
		return sbf.toString();
	}

    /**
     * 
     * @param dTime
     * @return
     */
    public String getItabName(String tabName,long dTime){
        Date dataTime = new Date(dTime);
        if(tabName.indexOf(CommParme.timeStartReg)!=-1 && tabName.indexOf(CommParme.timeEndReg)!=-1){
            String regStr = StringUtil.getMiddStr(tabName, CommParme.timeStartReg, CommParme.timeEndReg);
            tabName = tabName.replace(CommParme.timeStartReg+regStr+CommParme.timeEndReg, DateUtil.formatDateByFormat(dataTime, regStr,Locale.CHINA));
        }
        return tabName;
    }

    /* (non-Javadoc)
     * @see com.iqiyi.dbagent.contraller.i.IDataStoreHandler#dealData(java.io.BufferedReader, com.iqiyi.dbagent.base.config.i.Configuration)
     */
    public Object dealData(BufferedReader bfr, Configuration config) {
        // TODO Auto-generated method stub
        return null;
    }
}
